package com.atguigu.gmall.realtime.dwd.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;

import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {

    private Connection hBaseConn;
    private Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hBaseConn);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> tup2, Context context) throws Exception {
        com.alibaba.fastjson.JSONObject jsonObj = tup2.f0;
        TableProcessDim tableProcessDim = tup2.f1;
        //获取对业务数据库维度表操作类型
        String type = jsonObj.getString("type");
        jsonObj.remove("type");
        String sinkTable = tableProcessDim.getSinkTable();
        String rowKey = jsonObj.getString(tableProcessDim.getSinkRowKey());
        //判断
        if ("delete".equals(type)) {
            //说明对业务数据库维度表进行了删除操作，需要在Hbase对应的表中执行删除
            HBaseUtil.delRow(hBaseConn, Constant.HBASE_NAMESPACE,sinkTable,rowKey);
        } else {
            //说明对业务数据库维度表进行了添加或者修改，需要在HBase对应的表中执行put操作
            String family = tableProcessDim.getSinkFamily().split(",")[0];
            HBaseUtil.putRow(hBaseConn,Constant.HBASE_NAMESPACE,sinkTable,rowKey,family,jsonObj);
        }

        // 如果维度数据发生变化，将redis中缓存的维度数据清除掉
        if ("update".equals(type) || "delete".equals(type)) {
            String key = RedisUtil.getKey(sinkTable, rowKey);
            jedis.decr(key);
        }
    }
}
